/*
 * Copyright 1999-2020 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.nacos.common.remote.client;

import com.alibaba.nacos.api.ability.constant.AbilityKey;
import com.alibaba.nacos.api.ability.constant.AbilityStatus;
import com.alibaba.nacos.api.common.Constants;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.remote.RequestCallBack;
import com.alibaba.nacos.api.remote.RequestFuture;
import com.alibaba.nacos.api.remote.request.ClientDetectionRequest;
import com.alibaba.nacos.api.remote.request.ConnectResetRequest;
import com.alibaba.nacos.api.remote.request.HealthCheckRequest;
import com.alibaba.nacos.api.remote.request.Request;
import com.alibaba.nacos.api.remote.response.ClientDetectionResponse;
import com.alibaba.nacos.api.remote.response.ConnectResetResponse;
import com.alibaba.nacos.api.remote.response.ErrorResponse;
import com.alibaba.nacos.api.remote.response.Response;
import com.alibaba.nacos.common.executor.NameThreadFactory;
import com.alibaba.nacos.common.lifecycle.Closeable;
import com.alibaba.nacos.common.packagescan.resource.DefaultResourceLoader;
import com.alibaba.nacos.common.packagescan.resource.ResourceLoader;
import com.alibaba.nacos.common.remote.ConnectionType;
import com.alibaba.nacos.common.remote.PayloadRegistry;
import com.alibaba.nacos.common.utils.CollectionUtils;
import com.alibaba.nacos.common.utils.InternetAddressUtil;
import com.alibaba.nacos.common.utils.NumberUtils;
import com.alibaba.nacos.common.utils.StringUtils;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static com.alibaba.nacos.api.exception.NacosException.SERVER_ERROR;

/**
 * abstract remote client to connect to server.
 *
 * @author liuzunfei
 * @version $Id: RpcClient.java, v 0.1 2020年07月13日 9:15 PM liuzunfei Exp $
 */
@SuppressWarnings("PMD.AbstractClassShouldStartWithAbstractNamingRule")
@Slf4j
@Data
public abstract class RpcClient
        implements Closeable {

    private ServerListFactory serverListFactory;

    protected BlockingQueue<ConnectionEvent> eventLinkedBlockingQueue = new LinkedBlockingQueue<>();

    protected volatile AtomicReference<RpcClientStatus> rpcClientStatus = new AtomicReference<>(RpcClientStatus.WAIT_INIT);

    protected ScheduledExecutorService clientEventExecutor;

    private final ArrayBlockingQueue<ReconnectContext> reconnectionSignal = new ArrayBlockingQueue<>(1);

    protected volatile Connection currentConnection;

    private String tenant;

    private long lastActiveTimeStamp = System.currentTimeMillis();

    /**
     * listener called where connection's status changed.
     */
    protected List<ConnectionEventListener> connectionEventListeners = new ArrayList<>();

    /**
     * handlers to process server push request.
     */
    protected List<ServerRequestHandler> serverRequestHandlers = new ArrayList<>();

    private static final Pattern EXCLUDE_PROTOCOL_PATTERN = Pattern.compile("(?<=\\w{1,5}://)(.*)");

    protected RpcClientConfig rpcClientConfig;

    protected final ResourceLoader resourceLoader = new DefaultResourceLoader();

    static {
        PayloadRegistry.init();
    }

    public RpcClient(RpcClientConfig rpcClientConfig) {
        this(rpcClientConfig, null);
    }

    public RpcClient(RpcClientConfig rpcClientConfig,
                     ServerListFactory serverListFactory) {
        this.rpcClientConfig = rpcClientConfig;
        this.serverListFactory = serverListFactory;
        init();
    }

    protected void init() {
        if (this.serverListFactory != null) {
            rpcClientStatus.compareAndSet(RpcClientStatus.WAIT_INIT, RpcClientStatus.INITIALIZED);
            log.info("RpcClient init in constructor, ServerListFactory = {}", serverListFactory.getClass().getName());
        }
    }

    public RpcClient serverListFactory(ServerListFactory serverListFactory) {
        if (!isWaitInitiated()) {
            return this;
        }
        this.serverListFactory = serverListFactory;
        rpcClientStatus.compareAndSet(RpcClientStatus.WAIT_INIT, RpcClientStatus.INITIALIZED);

        log.info("[{}] RpcClient init, ServerListFactory = {}", rpcClientConfig.name(), serverListFactory.getClass().getName());
        return this;
    }

    protected void notifyDisConnected(Connection connection) {
        if (connectionEventListeners.isEmpty()) {
            return;
        }
        log.info("[{}] Notify disconnected event to listeners", rpcClientConfig.name());
        for (ConnectionEventListener connectionEventListener : connectionEventListeners) {
            try {
                connectionEventListener.onDisConnect(connection);
            } catch (Throwable throwable) {
                log.error("[{}] Notify disconnect listener error, listener = {}", rpcClientConfig.name(), connectionEventListener.getClass().getName());
            }
        }
    }

    protected void notifyConnected(Connection connection) {
        if (connectionEventListeners.isEmpty()) {
            return;
        }
        log.info("[{}] Notify connected event to listeners.", rpcClientConfig.name());
        for (ConnectionEventListener connectionEventListener : connectionEventListeners) {
            try {
                connectionEventListener.onConnected(connection);
            } catch (Throwable throwable) {
                log.error("[{}] Notify connect listener error, listener = {}", rpcClientConfig.name(), connectionEventListener.getClass().getName());
            }
        }
    }

    public boolean isWaitInitiated() {
        return this.rpcClientStatus.get() == RpcClientStatus.WAIT_INIT;
    }

    public boolean isRunning() {
        return this.rpcClientStatus.get() == RpcClientStatus.RUNNING;
    }

    public boolean isShutdown() {
        return this.rpcClientStatus.get() == RpcClientStatus.SHUTDOWN;
    }

    public void onServerListChange() {
        if (currentConnection != null && currentConnection.serverInfo != null) {
            ServerInfo serverInfo = currentConnection.serverInfo;
            boolean found = false;
            for (String serverAddress : serverListFactory.getServerList()) {
                if (resolveServerInfo(serverAddress).getAddress().equalsIgnoreCase(serverInfo.getAddress())) {
                    found = true;
                    break;
                }
            }
            if (!found) {
                log.info("Current connected server {} is not in latest server list, switch switchServerAsync", serverInfo.getAddress());
                switchServerAsync();
            }
        }
    }

    public final void start() throws NacosException {
        boolean success = rpcClientStatus.compareAndSet(RpcClientStatus.INITIALIZED, RpcClientStatus.STARTING);
        if (!success) {
            return;
        }
        clientEventExecutor = new ScheduledThreadPoolExecutor(2, new NameThreadFactory("com.alibaba.nacos.client.remote.worker"));
        clientEventExecutor.submit(() -> {
            while (!clientEventExecutor.isTerminated() && !clientEventExecutor.isShutdown()) {
                ConnectionEvent take;
                try {
                    take = eventLinkedBlockingQueue.take();
                    if (take.isConnected()) {
                        notifyConnected(take.connection);
                    } else if (take.isDisConnected()) {
                        notifyDisConnected(take.connection);
                    }
                } catch (Throwable e) {
                }
            }
        });

        clientEventExecutor.submit(() -> {
            while (true) {
                try {
                    if (isShutdown()) {
                        break;
                    }
                    ReconnectContext reconnectContext = reconnectionSignal.take();
                    //                    ReconnectContext reconnectContext = reconnectionSignal.poll(rpcClientConfig.connectionKeepAlive(), TimeUnit.MILLISECONDS);
                    if (reconnectContext == null) {
                        // check alive time.
                        if (System.currentTimeMillis() - lastActiveTimeStamp >= rpcClientConfig.connectionKeepAlive()) {
                            boolean isHealthy = healthCheck();
                            if (!isHealthy) {
                                if (currentConnection == null) {
                                    continue;
                                }
                                log.info("[{}] Server healthy check fail, currentConnection = {}", rpcClientConfig.name(), currentConnection.getConnectionId());

                                RpcClientStatus rpcClientStatus = RpcClient.this.rpcClientStatus.get();
                                if (RpcClientStatus.SHUTDOWN.equals(rpcClientStatus)) {
                                    break;
                                }

                                boolean statusFLowSuccess = RpcClient.this.rpcClientStatus.compareAndSet(rpcClientStatus, RpcClientStatus.UNHEALTHY);
                                if (statusFLowSuccess) {
                                    reconnectContext = new ReconnectContext(null, false);
                                } else {
                                    continue;
                                }
                            } else {
                                lastActiveTimeStamp = System.currentTimeMillis();
                                continue;
                            }
                        } else {
                            continue;
                        }
                    }

                    if (reconnectContext.serverInfo != null) {
                        boolean serverExist = false;
                        for (String server : getServerListFactory().getServerList()) {
                            ServerInfo serverInfo = resolveServerInfo(server);
                            if (serverInfo.getServerIp().equals(reconnectContext.serverInfo.getServerIp())) {
                                serverExist = true;
                                reconnectContext.serverInfo.serverPort = serverInfo.serverPort;
                                break;
                            }
                        }
                        if (!serverExist) {
                            log.info("[{}] Recommend server is not in server list, ignore recommend server {}", rpcClientConfig.name(), reconnectContext.serverInfo.getAddress());
                            reconnectContext.serverInfo = null;
                        }
                    }
                    reconnect(reconnectContext.serverInfo, reconnectContext.onRequestFail);
                } catch (Throwable throwable) {
                }
            }
        });

        Connection connectToServer = null;
        rpcClientStatus.set(RpcClientStatus.STARTING);

        int startUpRetryTimes = rpcClientConfig.retryTimes();
        while (startUpRetryTimes >= 0 && connectToServer == null) {
            try {
                startUpRetryTimes--;
                ServerInfo serverInfo = nextRpcServer();
                log.info("[{}] Try to connect to server on start up, server: {}", rpcClientConfig.name(), serverInfo);
                connectToServer = connectToServer(serverInfo);
            } catch (Throwable e) {
                log.warn("[{}] Fail to connect to server on start up, error message = {}, start up retry times left: {}", rpcClientConfig.name(), e.getMessage(), startUpRetryTimes, e);
            }
        }

        if (connectToServer != null) {
            log.info("[{}] Success to connect to server [{}] on start up, connectionId = {}", rpcClientConfig.name(), connectToServer.serverInfo.getAddress(), connectToServer.getConnectionId());
            this.currentConnection = connectToServer;
            rpcClientStatus.set(RpcClientStatus.RUNNING);
            eventLinkedBlockingQueue.offer(new ConnectionEvent(ConnectionEvent.CONNECTED, currentConnection));
        } else {
            switchServerAsync();
        }

        registerServerRequestHandler(new ConnectResetRequestHandler());
        registerServerRequestHandler((request, connection) -> {
            if (request instanceof ClientDetectionRequest) {
                return new ClientDetectionResponse();
            }
            return null;
        });

    }

    class ConnectResetRequestHandler
            implements ServerRequestHandler {
        @Override
        public Response requestReply(Request request,
                                     Connection connection) {
            if (request instanceof ConnectResetRequest) {
                try {
                    synchronized (RpcClient.this) {
                        if (isRunning()) {
                            ConnectResetRequest connectResetRequest = (ConnectResetRequest) request;
                            if (StringUtils.isNotBlank(connectResetRequest.getServerIp())) {
                                ServerInfo serverInfo = resolveServerInfo(connectResetRequest.getServerIp() + Constants.COLON + connectResetRequest.getServerPort());
                                switchServerAsync(serverInfo, false);
                            } else {
                                switchServerAsync();
                            }
                            afterReset(connectResetRequest);
                        }
                    }
                } catch (Exception e) {
                    log.error("[{}] Switch server error, {}", rpcClientConfig.name(), e);
                }
                return new ConnectResetResponse();
            }
            return null;
        }
    }

    protected void afterReset(ConnectResetRequest request) {
        // hook for GrpcClient
    }

    @Override
    public void shutdown() throws NacosException {
        log.info("Shutdown rpc client, set status to shutdown");
        rpcClientStatus.set(RpcClientStatus.SHUTDOWN);
        log.info("Shutdown client event executor " + clientEventExecutor);
        if (clientEventExecutor != null) {
            clientEventExecutor.shutdownNow();
        }
        closeConnection(currentConnection);
    }

    private boolean healthCheck() {
        HealthCheckRequest healthCheckRequest = new HealthCheckRequest();
        if (this.currentConnection == null) {
            return false;
        }
        int reTryTimes = rpcClientConfig.healthCheckRetryTimes();
        Random random = new Random();
        while (reTryTimes >= 0) {
            reTryTimes--;
            try {
                if (reTryTimes > 1) {
                    Thread.sleep(random.nextInt(500));
                }
                Response response = this.currentConnection.request(healthCheckRequest, rpcClientConfig.healthCheckTimeOut());
                // not only check server is ok, also check connection is register.
                return response != null && response.isSuccess();
            } catch (Exception e) {
                // ignore
            }
        }
        return false;
    }

    public void switchServerAsyncOnRequestFail() {
        switchServerAsync(null, true);
    }

    public void switchServerAsync() {
        switchServerAsync(null, false);
    }

    protected void switchServerAsync(final ServerInfo recommendServerInfo,
                                     boolean onRequestFail) {
        reconnectionSignal.offer(new ReconnectContext(recommendServerInfo, onRequestFail));
    }

    /**
     * switch server .
     */
    protected void reconnect(final ServerInfo recommendServerInfo,
                             boolean onRequestFail) {
        try {
            AtomicReference<ServerInfo> recommendServer = new AtomicReference<>(recommendServerInfo);
            if (onRequestFail && healthCheck()) {
                log.info("[{}] Server check success, currentServer is {} ", rpcClientConfig.name(), currentConnection.serverInfo.getAddress());
                rpcClientStatus.set(RpcClientStatus.RUNNING);
                return;
            }
            log.info("[{}] Try to reconnect to a new server, server is {}", rpcClientConfig.name(), recommendServerInfo == null
                                                                                                    ? " not appointed, will choose a random server."
                                                                                                    : (recommendServerInfo.getAddress() + ", will try it once."));
            boolean switchSuccess = false;
            int reConnectTimes = 0;
            int retryTurns = 0;
            Exception lastException;
            while (!switchSuccess && !isShutdown()) {
                // 1.get a new server
                ServerInfo serverInfo = null;
                try {
                    serverInfo = recommendServer.get() == null ? nextRpcServer() : recommendServer.get();
                    // 2.create a new channel to new server
                    Connection connectionNew = connectToServer(serverInfo);
                    if (connectionNew != null) {
                        log.info("[{}] Success to connect a server [{}], connectionId = {}", rpcClientConfig.name(), serverInfo.getAddress(), connectionNew.getConnectionId());
                        // successfully create a new connect.
                        if (currentConnection != null) {
                            log.info("[{}] Abandon prev connection, server is {}, connectionId is {}", rpcClientConfig.name(), currentConnection.serverInfo.getAddress(),
                                     currentConnection.getConnectionId());
                            // set current connection to enable connection event.
                            currentConnection.setAbandon(true);
                            closeConnection(currentConnection);
                        }
                        currentConnection = connectionNew;
                        rpcClientStatus.set(RpcClientStatus.RUNNING);
                        switchSuccess = true;
                        eventLinkedBlockingQueue.add(new ConnectionEvent(ConnectionEvent.CONNECTED, currentConnection));
                        return;
                    }

                    if (isShutdown()) {
                        closeConnection(currentConnection);
                    }

                    lastException = null;
                } catch (Exception e) {
                    lastException = e;
                } finally {
                    recommendServer.set(null);
                }

                if (CollectionUtils.isEmpty(RpcClient.this.serverListFactory.getServerList())) {
                    throw new Exception("server list is empty");
                }

                if (reConnectTimes > 0 && reConnectTimes % RpcClient.this.serverListFactory.getServerList().size() == 0) {
                    log.info("[{}] Fail to connect server, after trying {} times, last try server is {}, error = {}", rpcClientConfig.name(), reConnectTimes, serverInfo, lastException == null
                                                                                                                                                                          ? "unknown"
                                                                                                                                                                          : lastException);
                    if (Integer.MAX_VALUE == retryTurns) {
                        retryTurns = 50;
                    } else {
                        retryTurns++;
                    }
                }

                reConnectTimes++;

                try {
                    if (!isRunning()) {
                        Thread.sleep(Math.min(retryTurns + 1, 50) * 100L);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }

            if (isShutdown()) {
                log.info("[{}] Client is shutdown, stop reconnect to server", rpcClientConfig.name());
            }

        } catch (Exception e) {
            log.warn("[{}] Fail to reconnect to server, error is {}", rpcClientConfig.name(), e);
        }
    }

    private void closeConnection(Connection connection) {
        if (connection != null) {
            log.info("Close current connection " + connection.getConnectionId());
            connection.close();
            eventLinkedBlockingQueue.add(new ConnectionEvent(ConnectionEvent.DISCONNECTED, connection));
        }
    }

    public abstract ConnectionType getConnectionType();
    public abstract int rpcPortOffset();

    public ServerInfo getCurrentServer() {
        if (this.currentConnection != null) {
            return currentConnection.serverInfo;
        }
        return null;
    }

    public Response request(Request request) throws NacosException {
        return request(request, rpcClientConfig.timeOutMills());
    }

    public Response request(Request request,
                            long timeoutMills) throws NacosException {
        int retryTimes = 0;
        Response response;
        Throwable exceptionThrow = null;
        long start = System.currentTimeMillis();
        while (retryTimes <= rpcClientConfig.retryTimes() && (timeoutMills <= 0 || System.currentTimeMillis() < timeoutMills + start)) {
            boolean waitReconnect = false;
            try {
                if (this.currentConnection == null || !isRunning()) {
                    waitReconnect = true;
                    throw new NacosException(NacosException.CLIENT_DISCONNECT, "Client not connected, current status:" + rpcClientStatus.get());
                }
                response = this.currentConnection.request(request, timeoutMills);
                if (response == null) {
                    throw new NacosException(SERVER_ERROR, "Unknown Exception.");
                }
                if (response instanceof ErrorResponse) {
                    if (response.getErrorCode() == NacosException.UN_REGISTER) {
                        synchronized (this) {
                            waitReconnect = true;
                            if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) {
                                log.error("Connection is unregistered, switch server, connectionId = {}, request = {}", currentConnection.getConnectionId(), request.getClass().getSimpleName());
                                switchServerAsync();
                            }
                        }
                    }
                    throw new NacosException(response.getErrorCode(), response.getMessage());
                }
                lastActiveTimeStamp = System.currentTimeMillis();
                return response;
            } catch (Throwable e) {
                if (waitReconnect) {
                    try {
                        Thread.sleep(Math.min(100, timeoutMills / 3));
                    } catch (Exception exception) {
                    }
                }
                log.error("Send request fail, request = {}, retryTimes = {}, errorMessage = {}", request, retryTimes, e.getMessage());
                exceptionThrow = e;
            }
            retryTimes++;
        }

        if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) {
            switchServerAsyncOnRequestFail();
        }

        if (exceptionThrow != null) {
            throw (exceptionThrow instanceof NacosException) ? (NacosException) exceptionThrow : new NacosException(SERVER_ERROR, exceptionThrow);
        } else {
            throw new NacosException(SERVER_ERROR, "Request fail, unknown Error");
        }
    }

    public void asyncRequest(Request request,
                             RequestCallBack callback) throws NacosException {
        int retryTimes = 0;
        Throwable exceptionToThrow = null;
        long start = System.currentTimeMillis();
        while (retryTimes <= rpcClientConfig.retryTimes() && System.currentTimeMillis() < start + callback.getTimeout()) {
            boolean waitReconnect = false;
            try {
                if (this.currentConnection == null || !isRunning()) {
                    waitReconnect = true;
                    throw new NacosException(NacosException.CLIENT_DISCONNECT, "Client not connected.");
                }
                this.currentConnection.asyncRequest(request, callback);
                return;
            } catch (Throwable e) {
                if (waitReconnect) {
                    try {
                        // wait client to reconnect.
                        Thread.sleep(Math.min(100, callback.getTimeout() / 3));
                    } catch (Exception exception) {
                        // Do nothing.
                    }
                }
                log.error("[{}] Send request fail, request = {}, retryTimes = {}, errorMessage = {}", rpcClientConfig.name(), request, retryTimes, e.getMessage());
                exceptionToThrow = e;

            }
            retryTimes++;
        }

        if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) {
            switchServerAsyncOnRequestFail();
        }
        if (exceptionToThrow != null) {
            throw (exceptionToThrow instanceof NacosException) ? (NacosException) exceptionToThrow : new NacosException(SERVER_ERROR, exceptionToThrow);
        } else {
            throw new NacosException(SERVER_ERROR, "AsyncRequest fail, unknown error");
        }
    }

    public RequestFuture requestFuture(Request request) throws NacosException {
        int retryTimes = 0;
        long start = System.currentTimeMillis();
        Exception exceptionToThrow = null;
        while (retryTimes <= rpcClientConfig.retryTimes() && System.currentTimeMillis() < start + rpcClientConfig.timeOutMills()) {
            boolean waitReconnect = false;
            try {
                if (this.currentConnection == null || !isRunning()) {
                    waitReconnect = true;
                    throw new NacosException(NacosException.CLIENT_DISCONNECT, "Client not connected.");
                }
                return this.currentConnection.requestFuture(request);
            } catch (Exception e) {
                if (waitReconnect) {
                    try {
                        // wait client to reconnect.
                        Thread.sleep(100L);
                    } catch (Exception exception) {
                        // Do nothing.
                    }
                }
                log.error("[{}] Send request fail, request = {}, retryTimes = {}, errorMessage = {}", rpcClientConfig.name(), request, retryTimes, e.getMessage());
                exceptionToThrow = e;
            }
            retryTimes++;
        }

        if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) {
            switchServerAsyncOnRequestFail();
        }

        if (exceptionToThrow != null) {
            throw (exceptionToThrow instanceof NacosException) ? (NacosException) exceptionToThrow : new NacosException(SERVER_ERROR, exceptionToThrow);
        } else {
            throw new NacosException(SERVER_ERROR, "Request future fail, unknown error");
        }

    }

    public abstract Connection connectToServer(ServerInfo serverInfo) throws Exception;

    protected Response handleServerRequest(final Request request) {
        log.info("[{}] Receive server push request, request = {}, requestId = {}", rpcClientConfig.name(), request.getClass().getSimpleName(), request.getRequestId());
        lastActiveTimeStamp = System.currentTimeMillis();
        for (ServerRequestHandler serverRequestHandler : serverRequestHandlers) {
            try {
                Response response = serverRequestHandler.requestReply(request, currentConnection);

                if (response != null) {
                    log.info("[{}] Ack server push request, request = {}, requestId = {}", rpcClientConfig.name(), request.getClass().getSimpleName(), request.getRequestId());
                    return response;
                }
            } catch (Exception e) {
                log.info("[{}] HandleServerRequest:{}, errorMessage = {}", rpcClientConfig.name(), serverRequestHandler.getClass().getName(), e.getMessage());
                throw e;
            }
        }
        return null;
    }

    public synchronized void registerConnectionListener(ConnectionEventListener connectionEventListener) {

        log.info("[{}] Registry connection listener to current client:{}", rpcClientConfig.name(), connectionEventListener.getClass().getName());
        this.connectionEventListeners.add(connectionEventListener);
    }

    public synchronized void registerServerRequestHandler(ServerRequestHandler serverRequestHandler) {
        log.info("[{}] Register server push request handler:{}", rpcClientConfig.name(), serverRequestHandler.getClass().getName());
        this.serverRequestHandlers.add(serverRequestHandler);
    }

    public String getName() {
        return rpcClientConfig.name();
    }

    protected ServerInfo nextRpcServer() {
        String serverAddress = getServerListFactory().genNextServer();
        return resolveServerInfo(serverAddress);
    }

    protected ServerInfo currentRpcServer() {
        String serverAddress = getServerListFactory().getCurrentServer();
        return resolveServerInfo(serverAddress);
    }

    private ServerInfo resolveServerInfo(String serverAddress) {
        Matcher matcher = EXCLUDE_PROTOCOL_PATTERN.matcher(serverAddress);
        if (matcher.find()) {
            serverAddress = matcher.group(1);
        }
        String[] ipPortTuple = InternetAddressUtil.splitIPPortStr(serverAddress);
        int defaultPort = Integer.parseInt(System.getProperty("nacos.server.port", "8848"));
        String serverPort = CollectionUtils.getOrDefault(ipPortTuple, 1, Integer.toString(defaultPort));

        return new ServerInfo(ipPortTuple[0], NumberUtils.toInt(serverPort, defaultPort));
    }

    @AllArgsConstructor
    @NoArgsConstructor
    @Getter
    public static class ServerInfo {
        protected String serverIp;
        protected int serverPort;
        public String getAddress() {
            return serverIp + Constants.COLON + serverPort;
        }
        @Override
        public String toString() {
            return "{serverIp = '" + serverIp + '\'' + ", server main port = " + serverPort + '}';
        }
    }

    @AllArgsConstructor
    public static class ConnectionEvent {
        public static final int CONNECTED = 1;
        public static final int DISCONNECTED = 0;

        int eventType;
        Connection connection;

        public boolean isConnected() {
            return eventType == CONNECTED;
        }

        public boolean isDisConnected() {
            return eventType == DISCONNECTED;
        }
    }

    public Map<String, String> getLabels() {
        return rpcClientConfig.labels();
    }

    @AllArgsConstructor
    static class ReconnectContext {
        ServerInfo serverInfo;
        boolean onRequestFail;
    }

    public AbilityStatus getConnectionAbility(AbilityKey abilityKey) {
        if (currentConnection != null) {
            return currentConnection.getConnectionAbility(abilityKey);
        }
        return null;
    }
}
